package com.catmiao.spark.req

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @title: Spark01_Req1_HotCategoryTop10Analysis
 * @projectName spark_study
 * @description: 页面单挑转换率
 * @author ChengMiao
 * @date 2024/3/11 23:14
 */
object Spark03_Req1_PageflowAnalysis {

  def main(args: Array[String]): Unit = {


    // TODO 1. top10 热门品类
    val conf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10")

    val sc = new SparkContext(conf)


    // 1. 读取原始的日志数据
    val rdd = sc.textFile("datas/req_record/user_visit_action.txt")


    // 2. 数据转换实体
    val actionDataRdd: RDD[UserVisitAction] = rdd.map(
      item => {
        val datas = item.split("_")
        UserVisitAction(
          datas(0),
          datas(1).toLong,
          datas(2),
          datas(3).toLong,
          datas(4),
          datas(5),
          datas(6).toLong,
          datas(7).toLong,
          datas(8),
          datas(9),
          datas(10),
          datas(11),
          datas(12).toLong
        )
      }
    )

    actionDataRdd.cache()

    // 3. 计算分母 得到每个页面的点击次数即可
    val pageIds = List(1, 2, 3, 4, 5, 6, 7) // 过滤数据
    val pageIdsTuple: List[(Int, Int)] = pageIds.zip(pageIds.tail)
    val tuples: Array[(Long, Int)] = actionDataRdd.filter(
      item => {
        pageIds.contains(item.page_id)
      }
    ).map(
      item => {
        (item.page_id, 1)
      }
    ).reduceByKey(_ + _).collect()

    // 4. 计算分子
    // 4-1. 根据session_id 分组
    val sessionRdd: RDD[(String, Iterable[UserVisitAction])] = actionDataRdd.groupBy(_.session_id)
    // 4-1. 根据访问时间排序[升序]
    val mvRdd: RDD[(String, List[((Long, Long), Int)])] = sessionRdd.mapValues(
      iter => {
        val sortList = iter.toList.sortBy(_.action_time)
        // 排序后 只留下操作页面即可 [1,2,3,4]
        val flowIds: List[Long] = sortList.map(_.page_id)
        // 期望转换成 [1-2,2-3,3-4]
        // Sliding:滑窗  zip：拉链
        val pageFlowIds: List[(Long, Long)] = flowIds.zip(flowIds.tail)

        // 将不合法的页面跳转过滤
        pageFlowIds.filter(
          item=>{
            pageIdsTuple.contains(item)
          }
        ).map(
          tuple => {
            (tuple, 1)
          }
        )

      }
    )

    // ((1,2),1) 各页面路径的次数计数
    val flatRdd: RDD[((Long, Long), Int)] = mvRdd.map(_._2).flatMap(list => list)

    // 分子数据
    val dataRdd: RDD[((Long, Long), Int)] = flatRdd.reduceByKey(_ + _)

    // 分母转map
    val map: Map[Long, Int] = tuples.toMap

    // 计算单眺转换率 分子/分母
    dataRdd.foreach {
      case ((page1,page2),sum) => {
        val i = map.getOrElse(page1, 0)

          println(s"页面${page1} 跳转到 ${page2} 的单眺转换率:" + (sum.toDouble/i ))


      }
    }

    /**
     * 页面单挑转换率
     * - 定义：比如一个用户在一次Session过程中访问的页面路径 3,5,7,9,10,21，那么页面3跳到页面5叫一次单眺，7-9也叫一次单眺，单眺转换率就是要统计页面点击的概率
     */
    sc.stop()
  }


  /**
   * 样例类 用户访问动作表
   */
  case class UserVisitAction(
                              date: String, // 用户点击行为的日期
                              user_id: Long, // 用户id
                              session_id: String, // Session的id
                              page_id: Long, // 某个页面的id
                              action_time: String, // 动作的时间点
                              search_keyword: String, // 用户搜索的关键字
                              click_category_id: Long, // 某一个商品品类的id
                              click_product_id: Long, // 某一个商品的ID
                              order_category_ids: String, // 一次订单中所有品类的id集合
                              order_product_ids: String, // 一次订单中所有商品的id集合
                              pay_category_ids: String, // 一次支付中所有品类的id集合
                              pay_product_ids: String, // 一次支付中所有商品的id集合
                              city_id: Long // 城市id
                            )

}
